import time

import pandas as pd

from datetime import datetime
from math import fabs
import threading
from collections import defaultdict
from queue import Queue
from threading import Thread
from time import sleep
# from tkinter.messagebox import NO
from typing import Any, Callable, List
import json
import os
import yaml
import marshal
import redis
from redis import StrictRedis

# from tqsdk import TqApi, TqAuth, TqAccount, TqSim, __version__

host = "120.79.67.37"
port = 6379
password = "steward1!"
pool = redis.ConnectionPool(host=host, port=port,
                            db=3, password=password, decode_responses=True)
stra_rds = redis.StrictRedis(connection_pool=pool)
td_rds = redis.StrictRedis(connection_pool=redis.ConnectionPool(
    host=host, port=port, db=2, password=password, decode_responses=True))

INSTRUMENT = "inst."
FUTURE = "future."
OPTION = "option."
ORDER = "order."
PARKORDER = "park_order."
TRADE = "trade."
ACCOUNT = "acc."
POSITION = "position."
POSITIONDETAIL = "position_detail."
STRATEGY = "strategy."
STRASIGNAL = "eStraSignal."
TICK = "Tick."  # 后加 instrument id, publish 时的 channel
BAR = "Bar."

SUBSCRIBEINSTRUMENT = "Subscribe."
REQSUBINST = "ReqSubInst."
REQORDERINSERT = "ReqOrderInsert."
REQUNSUBINST = "ReqUnSubInst."
RQACC = "ReqQryAccount."
RQTRADE = "ReqQryTrade."
RQORDER = "ReqQryOrder."
RQPOSITION = "ReqQryPosition."
RQPOSITIONDETAIL = "ReqQryPositionDetail."

# # PUBLISH<=>SUBSCRIBE 用作channel
CTPWORK = "CTPWORK"  # msg:StartQuote,StopQuote,StartTrade,StopTrade
EVENT_MD_CONNECT_STATUS = "QuoteConnected"  # true or false  1 or 0
EVENT_MD_LOGIN_STATUS = "QuoteLogin"  # true or false
EVENT_TD_CONNECT_STATUS = "TradeConnected"
EVENT_TD_LOGIN_STATUS = "TradeLogin"
# # ctp redis 中数据库
COMMONDB = 0  # 通用数据库
QUOTEDB = 1  # 市场行情数据库
TRADEDB = 2  # 交易数据库
ACCOUNTDB = 10  # 账户数据库
STRATEGYDB = 3  # 策略数据库

EVENT_RQ = "rq_"  # 查询请求，可与其它联合使用：EVENT_RQ+EVENT_ORDER 生成 rq_eOrder.
EVENT_TIMER = "eTimer"  # noqa
EVENT_TICK = "eTick."
EVENT_TRADE = "eTrade."
EVENT_ORDER = "eOrder."
EVENT_PARK_ORDER = "eParkOrder."
EVENT_POSITION = "ePosition."
EVENT_ACCOUNT = "eAccount."
EVENT_CONTRACT = "eContract."
EVENT_LOG = "eLog"
EVENT_POSITION_DETAIL = "ePositionDetail."
EVENT_MD_LOGIN = "eMDLogin."
EVENT_TD_LOGIN = "eTDLogin."
EVENT_MD_CONNECTED = "eMDConnected."
EVENT_TD_CONNECTED = "eTDConnected."
EVENT_INSTRUMENT = "eInstrument."
EVENT_INSTRUMENT_OK = "eInstrumentOK."
EVENT_POSITION_DETAIL_OK = "ePositionDetailOK."
EVENT_POSITION_OK = "ePositionOK."
EVENT_INSTRUMENT_STATUS = "eInstrumentStatus."
EVENT_ACCOUNT_OK = "eAccountOK."
EVENT_ORDER_OK = "eOrderOK."
EVENT_FUTURE = "eFuture."
EVENT_STRA = "eStrategy."
EVENT_STRATEGY = "eStrategy."
EVENT_STRA_SIGNAL = "eStraSignal."
EVENT_STOP_ORDER = "eStopOrder."
EVENT_SUBSCRIBE = "eSubscribe."
EVENT_UNSUBSCRIBE = "eUnSubscribe."


class Event:
    """
    Event object consists of a type string which is used
    by event engine for distributing event, and a data
    object which contains the real data.
    """

    def __init__(self, event_type: str, data: Any = None):
        """"""
        self.event_type: str = event_type
        self.data: Any = data


# Defines handler function to be used in event engine.
HandlerType = Callable[[Event], None]


class RedisEventEngine:
    """
    Event engine distributes event object based on its type
    to those handlers registered.

    It also generates timer event by every interval seconds,
    which can be used for timing purpose.
    """

    def __init__(self, interval: int = 1, rds: StrictRedis = None):
        """
        Timer event is generated every 1 second by default, if
        interval not specified.
        """
        if rds is None:
            rds = td_rds
            # pool = redis.ConnectionPool(host=cfg.settings['redis']['host'], port=cfg.settings['redis']['port'], db=0,
            #                             decode_responses=True)
            # rds = redis.StrictRedis(connection_pool=pool)
        self._interval: int = interval
        self._rds: StrictRedis = rds

        self._active: bool = False
        self._thread: Thread = Thread(target=self._run)
        self._timer: Thread = Thread(target=self._run_timer)
        self._handlers: defaultdict = defaultdict(list)
        self._general_handlers: List = []
        self._listened_channels = []
        self._listener = []
        self.ps = self._rds.pubsub()  # redis.pubsub()
        # iid:[eid,pid,need_subscribed,subscribed,main_second(0,1,2)]
        self.iids = {}

        self._queue = Queue()

    def _run(self) -> None:
        """
        Get event from queue and then process it.
        """

        self.ps.subscribe('config')

        while self._active:

            for data in self.ps.listen():
                if data['type'] in ['message', 'pmessage']:
                    ed = data['data']
                    if isinstance(ed, str):
                        ed = json.loads(ed)
                    threading.Thread(target=self._process, args=(
                        Event(data['channel'], ed),)).start()

    def _process(self, event: Event) -> None:
        """
        First ditribute event to those handlers registered listening
        to this type.

        Then distrubute event to those general handlers which listens
        to all types.
        """
        if event.event_type in self._handlers:
            [handler(event) for handler in self._handlers[event.event_type]]

        if self._general_handlers:
            [handler(event) for handler in self._general_handlers]

    def _run_timer(self) -> None:
        """
        Sleep by interval second(s) and then generate a timer event.
        """
        while self._active:
            sleep(self._interval)
            event = Event(EVENT_TIMER)
            self.put(event)

    def put(self, event):
        self._queue.put(Event(event.event_type, event.data))

    def rds_put(self, event):
        self._rds.set(event.event_type, event.data)

    def start(self) -> None:
        """
        Start event engine to process events and generate timer events.
        """
        self._active = True
        self._thread.start()
        # self._timer.start()

    def stop(self) -> None:
        """
        Stop event engine.
        """
        self._active = False
        self._timer.join()
        self._thread.join()

    close = stop

    def publish(self, event: Event = None, event_type=None, data=None) -> None:
        """
        Publish an event object into redis server.
        """
        if event is None:
            self._rds.publish(event_type, json.dumps(data))
        else:
            # print(event.data)
            self._rds.publish(event.event_type, json.dumps(event.data))

    def register(self, event_type: str, handler: HandlerType) -> None:
        """
        Register a new handler function for a specific event type. Every
        function can only be registered once for each event type.
        """
        handler_list = self._handlers[event_type]
        if handler not in handler_list:
            handler_list.append(handler)
            self.ps.subscribe(event_type)

    def unregister(self, event_type: str, handler: HandlerType) -> None:
        """
        Unregister an existing handler function from event engine.
        """
        handler_list = self._handlers[event_type]

        if handler in handler_list:
            handler_list.remove(handler)

        if not handler_list:
            self._handlers.pop(event_type)
            self.ps.unsubscribe(event_type)

    def register_general(self, handler: HandlerType) -> None:
        """
        Register a new handler function for all event types. Every
        function can only be registered once for each event type.
        """
        if handler not in self._general_handlers:
            self._general_handlers.append(handler)

    def unregister_general(self, handler: HandlerType) -> None:
        """
        Unregister an existing general handler function.
        """
        if handler in self._general_handlers:
            self._general_handlers.remove(handler)


    def save_ss(self):
        """保存自选标的"""

    def load_ss(self):
        """载入自选标的"""

    def add2ss(self, iid):
        """加入自选"""

    def del_ss(self, iid):
        """从自选中删除"""

    def put_trade(self, data):
        put_ctp_trade_hash(data)

    def put_account(self, data):
        put_ctp_account_hash(data)

    def put_position(self, data):
        put_ctp_position_hash(data)

    def put_position_detail(self, data):
        put_ctp_position_detail_hash(data)

    def put_order(self, data):
        put_ctp_order_hash(data)

    def put_instrument(self, data):
        p_f, e_f = self.get_product_exchange_iid(data)
        self.instruments = data
        self.product_instrument = p_f
        self.exchange_instrument = e_f
        put_ctp_instrument_list(data)

    def put_future(self, data):
        # print(data)
        p_f, e_f = self.get_product_exchange_iid(data)
        self.futures = data
        self.product_future = p_f
        self.exchange_future = e_f
        put_ctp_future_list(data)

    def put_tick(self, data):
        put_ctp_tick(self._rds, data)

    def publish_tick(self, data):
        publish_ctp_tick(self._rds, data)

    def get_future(self):
        return get_ctp_future_list()

    def get_account(self):
        return get_ctp_account_hash()

    def get_order(self):
        return get_ctp_order_hash()

    def get_position(self):
        return get_ctp_position_hash()

    def get_position_detail(self):
        return get_ctp_position_detail_hash()

    def get_trade(self):
        return get_ctp_trade_hash()


"""/*begin region of redis operation on CTP*/
"""


def put_ctp_instrument_list(data, rds=td_rds):
    rds.rpush(EVENT_INSTRUMENT, json.dumps(data))


def get_ctp_instrument_list(rds=td_rds):
    data = rds.lrange(EVENT_INSTRUMENT, 0, -1)
    if not data:
        return

    return json.loads(data[0])


def get_ctp_future_list(rds=td_rds):
    data = rds.lrange(EVENT_FUTURE, 0, -1)

    if not data:
        return

    return json.loads(data[0])


def get_ctp_account_list(rds=td_rds, acc_id=''):
    data = rds.lrange(EVENT_ACCOUNT + acc_id, 0, -1)
    if not data:
        return None
    return json.loads(data[0])


def get_ctp_position_list(rds=td_rds, acc_id=''):
    data = rds.lrange(EVENT_POSITION + acc_id, 0, -1)

    if not data:
        return

    return [json.loads(data[i]) for i in range(len(data))]


def get_ctp_position_detail_list(rds=td_rds, acc_id=''):
    data = rds.lrange(EVENT_POSITION_DETAIL + acc_id, 0, -1)

    if not data:
        return

    return [json.loads(data[i]) for i in range(len(data))]


def get_ctp_trade_list(rds=td_rds, acc_id=''):
    data = rds.lrange(EVENT_TRADE + acc_id, 0, -1)

    if not data:
        return

    return [json.loads(data[i]) for i in range(len(data))]


def get_ctp_order_list(rds=td_rds, acc_id=''):
    data = rds.lrange(EVENT_ORDER + acc_id, 0, -1)

    if not data:
        return

    return [json.loads(data[i]) for i in range(len(data))]


def put_ctp_future_list(data, rds=td_rds):
    # print(data)
    rds.rpush(EVENT_FUTURE, json.dumps(data))
    # save_yml('future', data)


def put_ctp_account_list(data, rds=td_rds, acc_id=''):
    rds.rpush(EVENT_ACCOUNT + acc_id, json.dumps(data))


def put_ctp_position_list(data, rds=td_rds, acc_id=''):
    rds.rpush(EVENT_POSITION + acc_id, json.dumps(data))


def put_ctp_position_detail_list(data, rds=td_rds, acc_id=''):
    rds.rpush(EVENT_POSITION_DETAIL + acc_id, json.dumps(data))


def put_ctp_trade_list(data, rds=td_rds, acc_id=''):
    rds.rpush(EVENT_TRADE + acc_id, json.dumps(data))


def put_ctp_order_list(data, rds=td_rds, acc_id=''):
    rds.rpush(EVENT_ORDER + acc_id, json.dumps(data))


def put_config(rds, data):
    rds.rpush("Config", json.dumps(data))


def put_ctp_tick(rds, data):
    rds.rpush(EVENT_TICK, json.dumps(data))


def publish_ctp_tick(rds, data):
    rds.publish(EVENT_TICK + data['InstrumentID'], json.dumps(data))


def put_hash(rds, name, *args, **kwargs):
    # print(f'{name}:\n{args}\n{kwargs}')
    if len(args) == 2:
        rds.hset(name, args)
    else:
        rds.hmset(name, kwargs)


def get_hash(rds, name, key=None):
    data = None

    if key and isinstance(key, str):
        data = rds.hget(name, key)
    elif key and isinstance(key, list):
        data = rds.hmget(name, key)
    else:
        data = rds.hgetall(name)
    # if name == EVENT_POSITION_DETAIL:
    # print(data)
    return {k: json.loads(data[k]) for k in data.keys()}


def put_ctp_account_hash(data, rds=td_rds, acc_id='', name=EVENT_ACCOUNT):
    acc_id = data.get('AccID', 'AccID') if not acc_id else acc_id
    rds.hset(name, acc_id, json.dumps(data))


def get_ctp_account_hash(rds=td_rds, acc_id='', name=EVENT_ACCOUNT):
    data = get_hash(rds, name, acc_id)
    if data:
        return data


def put_ctp_position_hash(data, rds=td_rds, pid='', name=EVENT_POSITION, acc_id=''):
    # print(f'put_ctp_position_hash:\n{data}')
    for k, v in data.items():
        rds.hset(name + acc_id, k, json.dumps(v))


def get_ctp_position_hash(rds=td_rds, pid='', name=EVENT_POSITION, acc_id=''):
    return get_hash(rds, name + acc_id, pid)


def put_ctp_position_detail_hash(data, rds=td_rds, pid='', name=EVENT_POSITION_DETAIL, acc_id=''):
    # print(data)
    rds.hset(name + acc_id, pid, json.dumps(data))


def get_ctp_position_detail_hash(rds=td_rds, pid='', name=EVENT_POSITION_DETAIL, acc_id=''):
    return get_hash(rds, name + acc_id, pid)


def put_ctp_order_hash(data, rds=td_rds, oid='', name=EVENT_ORDER, acc_id=''):
    if data:
        oid = data.get('OrderID', oid)
    rds.hset(name + acc_id, oid, json.dumps(data))


def get_ctp_order_hash(rds=td_rds, oid='', name=EVENT_ORDER):
    return get_hash(rds, name, oid)


def put_ctp_trade_hash(data, rds=td_rds, tid='', name=EVENT_TRADE, acc_id=''):
    if not tid:
        tid = data.get('TradeID', 'TradeID')
    rds.hset(name + acc_id, tid, json.dumps(data))


def get_ctp_trade_hash(rds=td_rds, tid='', name=EVENT_TRADE, acc_id=''):
    return get_hash(rds, name + acc_id, tid)


def put_stra_hash(stra_instance_id, rds=stra_rds, name=EVENT_STRA):
    rds.hset(name, stra_instance_id, 1)


def put_stra_signal_hash(stra_signal, stra_instance_id, rds=stra_rds, name=EVENT_STRA):
    rds.hset(name, stra_instance_id, stra_signal)


def publish_stra(stra_instance_id, rds=stra_rds):
    rds.publish(EVENT_STRA, stra_instance_id)


def publish_stra_signal(stra_instance_id, stra_signal, rds=stra_rds):
    rds.publish(EVENT_STRA + stra_instance_id, json.dumps(stra_signal))



"""
可以直接使用的期货公司：
G国际期货
H华安期货
G广发期货

第一列为能够直接使用的期货公司列表

第二列为可以使用天勤量化,但需要额外跟期货公司的客户经理申请使用的列表

第三列为可以使用天勤量化来进行ETF期权交易的期货公司，但是需配合完成ETF期权开户和程序化申请步骤

可直接使用	申请后才能使用	支持ETF期权
A安粮期货	F方正中期	W五矿经易_ETF
B倍特期货	H弘业期货	N南华期货_ETF
B宝城期货	Y银河期货	G国泰君安_ETF
B渤海期货	G国联期货	G光大期货_ETF
C创元期货	D东证期货	G国元期货_ETF
C长城期货	G光大期货	H宏源期货_ETF
C长安期货	S申万期货	
C长江期货	H华泰期货	
D东华期货	D东航期货	
D东方汇金	C长安期货	
D东海期货	Z中金岭南期货	
D东航期货	G国泰君安	
D东证期货		
D大地期货		
D大有期货		
D大越期货		
D大陆期货		
F方正中期		
F福能期货		
G光大期货		
G冠通期货		
G国信期货		
G国元期货		
G国富期货		
G国投安信		
G国新国证期货		
G国海良时		
G国盛期货		
G国联期货		
G国贸期货		
G国金期货		
G国际期货		
G广发期货		
G广金期货		
G格林大华		
G港信期货		
H华创期货		
H华安期货		
H华泰期货		
H华金期货		
H华鑫期货		
H华闻期货		
H华龙期货		
H和合期货		
H和融期货		
H宏源期货		
H弘业期货		
H徽商期货		
H恒力期货		
H恒泰期货		
H海航期货		
H海证期货		
H海通期货		
H混沌天成		
H红塔期货		
J建信期货		
J江海汇鑫		
J金信期货		
J金元期货		
J金瑞期货		
J金石期货		
J锦泰期货		
L鲁证期货(中泰期货)		
M民生期货		
M迈科期货		
N南华期货		
P平安期货		
Q前海期货		
Q齐盛期货		
R瑞奇期货		
R瑞达期货		
S上海东亚		
S上海东方		
S上海中期		
S山西三立		
S山金期货		
S申万期货		
S盛达期货		
S神华期货		
T通惠期货		
T铜冠金源		
W五矿期货		
X信达期货		
X先融期货		
X先锋期货		
X兴业期货		
X兴证期货		
X新世纪期货		
X新湖期货		
X新纪元期货		
X西南期货		
X西部期货		
Y一德期货		
Y英大期货		
Y银河期货		
Z中信建投		
Z中信期货		
Z中原期货		
Z中天期货		
Z中州期货		
Z中泰期货_数讯程序化		
Z中粮期货		
Z中航期货		
Z中融汇信		
Z中衍期货		
Z中财期货		
Z中辉期货		
Z中金财富		
Z中钢期货		
Z中银期货		
Z招商期货		
Z浙商期货		
Z紫金天风
"""


def output(event: Event):
    print('output', event.data)


def output1(event):
    print('output1(event):output2', event.event_type)


if __name__ == "__main__":
    ee = RedisEventEngine()
    ee.start()
    ee.register('output', output)
    # sleep(2)
    ee.register('output1', output1)
    # sleep(2)
    ee.publish(Event('output', 12345))
    # sleep(2)
    ee.publish(Event('output1', 44444))
    ee.publish(Event('output', '55555'))
